package cloud.xiguapi.ubas.analysis.hotitems;

import cloud.xiguapi.ubas.analysis.hotitems.model.UserBehaviorAscendingTimestampExtractor;
import cloud.xiguapi.ubas.model.UserBehavior;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Slide;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * @author 大大大西西瓜皮🍉
 * date: 2021-5-18 下午 08:25
 * desc:
 */
@Component
public class SqlHotItemsAnalysis {

    @Resource(name = "env")
    private StreamExecutionEnvironment env;

    @Async("sqlHotItemsAnalysis")
    public void analysis(String path) throws Exception {
        env.setParallelism(8);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // 读取数据源
        DataStream<String> inputDataStream = env.readTextFile(path);

        // 转换成Java Bean
        DataStream<UserBehavior> userBehaviorDataStream = inputDataStream.map(line -> {
            String[] fields = line.split(",");
            return UserBehavior.builder()
                    .userId(Long.valueOf(fields[0]))
                    .itemId(Long.valueOf(fields[1]))
                    .categoryId(Long.valueOf(fields[2]))
                    .behavior(fields[3])
                    .timestamp(Long.valueOf(fields[4]))
                    .build();
        }).assignTimestampsAndWatermarks(new UserBehaviorAscendingTimestampExtractor());

        // 创建表执行环境(Blink版本)
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        // 将流转换成表, 过滤不必要的字段
        Table dataTable = tableEnv.fromDataStream(userBehaviorDataStream, "itemId, behavior, timestamp.rowtime as ts");

        // 分组开窗
        Table windowAggregateTable = dataTable
                .filter("behavior = 'pv'")
                .window(Slide.over("1.hours").every("5.minutes").on("ts").as("w"))
                .groupBy("itemId, w")
                .select("itemId, w.end as windowEnd, itemId.count as cnt");

        // 利用开窗函数, 对count值进行排序并获取Row number, 得到Top N  -- (使用SQL实现)
        DataStream<Row> aggregateDataStream = tableEnv.toAppendStream(windowAggregateTable, Row.class);
        tableEnv.createTemporaryView("agg", aggregateDataStream, "itemId, windowEnd, cnt");

        // 对聚合后的结果集进行合并和筛选top n
        Table resultTable = tableEnv.sqlQuery("select * from " +
                " ( select *, ROW_NUMBER() over (partition by windowEnd order by cnt desc) as row_num from agg ) " +
                " where row_num <= 5");

        // 纯SQL实现
        tableEnv.createTemporaryView("data_table", userBehaviorDataStream, "itemId, behavior, timestamp.rowtime as ts");
        Table resultSqlTable = tableEnv.sqlQuery("select * from " +
                " ( select *, ROW_NUMBER() over (partition by windowEnd order by cnt desc) as row_num from (" +
                " select itemId, count(itemId) as cnt, HOP_END(ts, interval '5' minute, interval '1' hour) as windowEnd " +
                " from data_table where behavior = 'pv' group by itemId, HOP(ts, interval '5' minute, interval '1' hour)" +
                ") ) " +
                " where row_num <= 5");

        tableEnv.toRetractStream(resultTable, Row.class);
        tableEnv.toRetractStream(resultSqlTable, Row.class).print();

        env.execute("sql-hot-items");
    }
}
